package com.atguigu.flink.keyStateAndOpeState;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

public class OperatorListState {
    public static void main(String[] args) throws Exception {
        // TODO 1.环境准备
        // 1.1指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2设置并行度
        env.setParallelism(2);
        env.enableCheckpointing(10000);
        // TODO 2.从指定的网络端口读取数据
        DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 8888);
        // TODO 3.统计
        SingleOutputStreamOperator<String> processDS = socketDS.process(new MyPro());
        // TODO 6.打印输出
        processDS.print();
        // TODO 7.提交作业
        env.execute();

    }
}

class MyPro extends ProcessFunction<String , String> implements CheckpointedFunction {

    Integer count = 0;

    ListState<Integer> countStatus;

    @Override
    public void processElement(String value, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {
        out.collect("当前并行子任务" + getRuntimeContext().getIndexOfThisSubtask() + "处理的数据条数" + ++count);
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        System.out.println("snapshotState");
        countStatus.clear();
        countStatus.add(count);
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        System.out.println("initializeState");
        //注意：算子状态的初始化过程和键控状态不一样
        //键控状态：open         算子状态：initializeState
        //countState = getRuntimeContext().getListState()
        ListStateDescriptor<Integer> listStateDescriptor = new ListStateDescriptor<>("", Integer.class);
        countStatus = context.getOperatorStateStore().getListState(listStateDescriptor);
        if (context.isRestored()) {
            count = countStatus.get().iterator().next();
        }
    }
}
